//package com.css.jyzgl.kafka;
//
//import com.css.jyzgl.constants.Constants;
//import com.css.jyzgl.entity.dto.JyzjylsjlSyncRequest;
//import com.css.jyzgl.entity.dto.YjyqDTO;
//import com.css.jyzgl.service.JyzJyskJyjlService;
//import com.css.jyzgl.service.YqypGlService;
//import com.css.util.RedisUtils;
//import com.fasterxml.jackson.core.JsonProcessingException;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.google.gson.Gson;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource;
//import java.math.BigDecimal;
//import java.util.HashMap;
//import java.util.Map;
//
///**
//  * @Description: 加油记录消费者
//  * @Autor wujy
//  * @Date 2025/7/3
//  */
//@Slf4j
//@Component
//public class RefuelRecordConsumer {
//    @Resource
//    private JyzJyskJyjlService jyjlService;
//    @Resource
//    private YqypGlService yqyqService;
//
//    private static ObjectMapper objectMapper = new ObjectMapper();
//
//    @KafkaListener(topics = {Constants.JYZ_ZFKP_REFUEL_RECORD})
//    public void recordConsumer(ConsumerRecord<String, Object> record) throws JsonProcessingException {
//        log.info("refuel record consumer receive a message, partition {} ,offset {}", record.partition(), record.offset());
//        JyzjylsjlSyncRequest refuelRecord = getObjectMapper().readValue(record.value().toString(), JyzjylsjlSyncRequest.class);
//        String oilSpecKey = "oil_spec", unityPriceKey = "unity_price";
//        Map<String,Object> latestPrice = new HashMap<>();
//        latestPrice.put(oilSpecKey,refuelRecord.getOilType());
//        latestPrice.put(unityPriceKey, refuelRecord.getUnitPrice().toString());
//        String key = String.format(Constants.JYZ_LATEST_PRICE_KEY, refuelRecord.getCreditCode(),refuelRecord.getTankerCode(),refuelRecord.getGunPort());
//        String oilSpec = RedisUtils.getHash(key, oilSpecKey);
//        if (StringUtils.isNotBlank(oilSpec)) {
//            //比较油品油价
//            Double unityPrice = new Double(RedisUtils.getHash(key, unityPriceKey));
//            if (!StringUtils.equals(refuelRecord.getOilType(), oilSpec) || refuelRecord.getUnitPrice().doubleValue() != unityPrice) {
//                log.info("{} 油枪信息已变更",key);
//                //更新缓存信息
//                RedisUtils.putHash(key, latestPrice);
//                //更新油机油枪
//                YjyqDTO yjyqDTO = new YjyqDTO();
//                yjyqDTO.setNsrsbh(refuelRecord.getCreditCode());
//                yjyqDTO.setFueller(refuelRecord.getTankerCode());
//                yjyqDTO.setFuelGan(refuelRecord.getGunPort());
//                yjyqDTO.setOilSpec(refuelRecord.getOilType());
//                yjyqDTO.setUnitPrice(refuelRecord.getUnitPrice());
//                yqyqService.updOilSpecUnitPrice(yjyqDTO);
//            }
//        } else{
//            //保存最新价格
//            RedisUtils.putHash(key, latestPrice);
//        }
//
//        jyjlService.receiveRecord(refuelRecord);
//    }
//
//
//    public ObjectMapper getObjectMapper() {
//        if (null == objectMapper) {
//            objectMapper = new ObjectMapper();
//        }
//        return objectMapper;
//    }
//
//}
